package com.atguigu.dataStreamApi.sink;

import com.atguigu.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import util.SourceUtil;

import java.sql.PreparedStatement;
import java.sql.SQLException;

public class Flink04_JdbcSink {
      public static void main(String[] args) {
              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
              env.setParallelism(1);

              //开启检查点,如果开启批传可以不打开检查点
            env.enableCheckpointing(2000L);

          //对接数据源
          DataStreamSource<Event> ds =
                  env.fromSource(SourceUtil.getSource(), WatermarkStrategy.noWatermarks(), "dataGenSource");
          //将数据写到mysql表中
          SinkFunction<Event> sinkFunction = JdbcSink.<Event>sink(
                  //"insert into sink_to_mysql_test values(? ,? ,?)",//所建表没有设置主键约束，可以一直写入
                  //"insert into sink_to_mysql_test1 values(? ,? ,?)",//所建表设置主键约束，开启检查点时重复数据输入会报错。因此可通过下面这个语法解决
                 //todo 幂等写入
                  "replace into sink_to_mysql_test1 values(? ,? ,?)",//1.将除主键外的所有字段更行
                 // "insert into sink_to_mysql_test1 values(? ,? ,?) duplicate key update url = VALUES(url)",//2.将制定字段更新
                  new JdbcStatementBuilder<Event>() {
                      @Override
                      public void accept(PreparedStatement preparedStatement, Event event) throws SQLException {
                          //给sql占位符
                          preparedStatement.setString(1,event.getUser());
                          preparedStatement.setString(2, event.getUrl());
                          preparedStatement.setLong(3,event.getTs());
                      }
                  },
                  new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                          .withDriverName("com.mysql.cj.jdbc.Driver")
                          .withUrl("jdbc:mysql://hadoop102:3306/test")
                          .withUsername("root")
                          .withPassword("000000")
                          .build()
          );

          ds.addSink( sinkFunction );

          try {
                  env.execute();
              } catch (Exception e) {
                  throw new RuntimeException(e);
              }
          }
}
